﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace PerfTestLib.Utility.Parallel
{
    /// <summary>
    /// Ensure that parallel operations run under certain conditions.
    /// </summary>
    public class EnsureParallel
    {
        // degree of parallelism that need to be ensured
        private int _degreeOfParallelism;

        // semaphore for ensuring the max number of running threads
        private SemaphoreSlim _sem;

        // cancellation token
        private CancellationToken _cancellationToken;

        // barrier for ensuring the correct number of parallel threads
        private Barrier _ensureParallelCountBarrier;

        // determine whether the barrier is enabled
        private bool _ensureParallelCountBarrierEnabled = true;

        // list of exceptions thrown during the execution
        private ConcurrentBag<Exception> _innerExLst = new ConcurrentBag<Exception>();

        // dictionary of currently running threads handles
        private ConcurrentDictionary<int, ManualResetEvent> _threads = new ConcurrentDictionary<int, ManualResetEvent>();


        /// <summary>
        /// Ctor.
        /// </summary>
        /// <param name="degreeOfParallelism"></param>
        /// <param name="cancellationToken"></param>
        public EnsureParallel(int degreeOfParallelism, CancellationToken cancellationToken)
        {
            _degreeOfParallelism = degreeOfParallelism;
            _sem = new SemaphoreSlim(degreeOfParallelism);
            _cancellationToken = cancellationToken;
            _ensureParallelCountBarrier = new Barrier(_degreeOfParallelism);
        }

        /// <summary>
        /// Queue work item.
        /// </summary>
        /// <param name="action"></param>
        public void QueueUserWorkItem(Action action)
        {
            // +1 semaphore
            _sem.Wait(_cancellationToken);

            // check cancellation
            if (_cancellationToken.IsCancellationRequested)
                return;

            bool local_ensureParallelCountBarrierEnabled = _ensureParallelCountBarrierEnabled;

            // current thread ID
            int threadId = Thread.CurrentThread.ManagedThreadId;

            // wait handle to control the end of thread
            ManualResetEvent endOfThreadWaitHandle = new ManualResetEvent(false);

            // add thread id to list of currently running threads                        
            this._threads.TryAdd(threadId, endOfThreadWaitHandle);

            // run inside one thread
            ThreadPool.QueueUserWorkItem(obj =>
                {
                    try
                    {                        
                        // if parallel count should be ensured then check barrier
                        if (local_ensureParallelCountBarrierEnabled)
                        {
                            // + 1 barrier
                            _ensureParallelCountBarrier.SignalAndWait(_cancellationToken);
                        }

                        // check cabcellation
                        if (_cancellationToken.IsCancellationRequested)
                            return;                        

                        // execute action
                        action();
                    }
                    catch(Exception ex)
                    {
                        // add exception to list
                        _innerExLst.Add(ex);
                    }
                    finally
                    {
                        // -1 semaphore
                        _sem.Release();

                        // remove thread from the list of currently active threads
                        ManualResetEvent wh;
                        this._threads.TryRemove(threadId, out wh);

                        // mark thread as completed
                        endOfThreadWaitHandle.Set();
                    }
                });
        }

        /// <summary>
        /// Enable/disable ensuring parallel count.
        /// </summary>
        /// <param name="enabled"></param>
        public void EnableDisableEnsureParallelCount(bool enabled)
        {
            this._ensureParallelCountBarrierEnabled = enabled;
        }

        /// <summary>
        /// Wait for all threads to finish.
        /// </summary>
        public void Wait()
        {
            _threads.ToList().ForEach(th =>
                {
                    th.Value.WaitOne();
                });
        }

        public static void RunFor(int fromInclusive, int toExclusive, int degreeOfParallelism, CancellationToken cancellationToken, Action<int> body)
        {
            EnsureParallel parallel = new EnsureParallel(degreeOfParallelism, cancellationToken);

            for (int index = fromInclusive; index < toExclusive; index++)
            {
                if (parallel._innerExLst.Count > 0)
                {
                    break;
                }

                // if the remaining items are less than the degree of parallelism then disable ensuring parallel count
                if (toExclusive + 1 - index < degreeOfParallelism)
                {
                    parallel.EnableDisableEnsureParallelCount(false);
                }

                // run action in parallel
                parallel.QueueUserWorkItem(() =>
                    {
                        body(index);
                    });
            }

            parallel.Wait();

            if (parallel._innerExLst.Count > 0)
            {
                throw new AggregateException(parallel._innerExLst.ToList());
            }
        }
    }
}
